package sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Base64;


/**
 3.3 连接器属性
 属性	是否必填	默认值	类型 描述
 connector	required	(none)	String	‘datagen’
 rows-per-second	optional	10000 1000010000	Long	数据生产速率
 number-of-rows	optional	(none)	Long	指定生产的数据条数，默认是不限制
 fields.#.kind	optional	random	String	指定字段的生产数据的方式 random 还是 sequence
 fields.#.min	optional	(Minimum value of type)	(Type of field)	random 生成器的指定字段 # 最小值，支持数字类型
 fields.#.max	optional	(Maximum value of type)	(Type of field)	random 生成器的指定字段 # 最大值，支持数字类型
 fields.#.length	optional	100 100100	Integer	char / varchar / string / array / map / multiset 类型的长度
 fields.#.start	optional	(none)	(Type of field)	sequence 生成器的开始值
 fields.#.end	optional	(none)	(Type of field)	sequence 生成器的结束值
 */
public class D1Gen {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
                " stime TIMESTAMP(3)," +
                " name STRING," +
                " id BIGINT," +
                " ts AS localtimestamp," +
                " rowtime AS localtimestamp," +
                " WATERMARK FOR ts AS ts - interval '30' second" +
                ") WITH ( " +
                " 'connector'='datagen', " +
                " 'rows-per-second'='1'," +
                " 'fields.id.kind'='sequence'," +
                " 'fields.id.start'='1'," +
                " 'fields.id.end'='10000000'" +
                ")";


        String sink = "CREATE TABLE pprint (" +
                "    stime  TIMESTAMP(3)," +
                "    name STRING, " +
                "    id BIGINT " +
                ") WITH (" +
                "    'connector' = 'print'" +
                ")";
        String sinkSql = "INSERT INTO pprint " +
                "SELECT " +
                "stime, name, id " +
                " FROM ods_tb";

        String sql = genSql + ";\n" + sinkSql + ";\n" + sink + ";\n" + "\u0001" + "job.name=sql-test\n" +
                "job.parallelism=3\n" +
                "job.max.parallelism=9\n" +
                "job.enable.checkpointing.interval=60\n" +
                "job.checkpointing.mode=exactly-once\n" +
                "job.state.backend=rocksdb\n" +
                "job.state.backend.uri=hdfs:///user/flink/calc_primary/flink-checkpoints\n" +
                "job.checkpoint.cleanup.strategy=delete\n" +
                "job.checkpoint.parallelism=1\n" +
                "job.enable.checkpoint.unaligned.enable=false\n" +
                "job.checkpoint.timeout.ms=120000\n" +
                "job.checkpoint.failure.numbers=10\n" +
                "job.restart.strategy=fixed-delay\n" +
                "job.restart.attempts=1000\n" +
                "job.restart.delay.interval=30";

        String encode = Base64.getEncoder().encodeToString(sql.getBytes());
        System.out.println(encode);


        tableEnv.executeSql(genSql);
        tableEnv.executeSql(sink);
        tableEnv.executeSql(sinkSql);




    }
}
